package com.sh.data.engine.bootstrap;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.google.common.collect.Lists;
import com.sh.data.engine.common.enumDefinition.DSType;
import com.sh.data.engine.domain.integration.online.model.param.StartOnlineKafkaTaskParam;
import com.sh.data.engine.domain.integration.online.service.FlinkDataSyncService;
import com.sh.data.engine.job.admin.core.model.XxlJobGroup;
import com.sh.data.engine.job.admin.dao.XxlJobGroupDao;
import com.sh.data.engine.repository.dao.base.EngineBaseModel;
import com.sh.data.engine.repository.dao.integration.online.entity.FlinkDataSyncTaskEntity;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.List;

/**
 * @author: mengzheng.mhc
 * @date: 2025/9/17 14:15
 */
@Slf4j
@Component
public class AppInitializerRunner implements CommandLineRunner {

    @Autowired
    private XxlJobGroupDao xxlJobGroupDao;

    @Autowired
    private FlinkDataSyncService flinkDataSyncService;

    @Override
    public void run(String... args) throws Exception {
        try {
            doAddXxlGroup();
        } catch (Exception e) {
            log.error("更新xxl job信息失败", e);
        }
        log.info("更新xxl job信息成功");

        // 重启实时同步任务(非flink任务)
        try {
            restartOnlineTask();
        } catch (Exception e) {
            log.error("重启实时同步任务失败", e);
        }
        log.info("重启实时同步任务成功");
    }

    private void doAddXxlGroup() {
        XxlJobGroup xxlJobGroup = xxlJobGroupDao.selectById(2);
        String localHost = null;
        try {
            InetAddress inetAddress = InetAddress.getLocalHost();
            localHost = inetAddress.getHostAddress();
            localHost = "http://" + localHost + ":9999/";
        } catch (UnknownHostException e) {
            log.info("获取本地ip失败", e);
            XxlJobGroup xxlJobGroup1 = xxlJobGroupDao.selectById(1);
            localHost = xxlJobGroup1.getAddressList();
        }
        xxlJobGroup.setAddressList(localHost);
        xxlJobGroupDao.updateById(xxlJobGroup);
    }

    private void restartOnlineTask() {

        /** 查询所有已发布任务 */
        List<FlinkDataSyncTaskEntity> list =
            flinkDataSyncService.list(
                new LambdaQueryWrapper<>(FlinkDataSyncTaskEntity.class)
                    .eq(FlinkDataSyncTaskEntity::getTaskStatus, 2)
                    .in(
                        FlinkDataSyncTaskEntity::getSourceDsType,
                        Lists.newArrayList(DSType.Kafka.name(), DSType.MQTT.name())));

        List<Long> collect = list.stream().map(EngineBaseModel::getId).toList();

        // 先取消发布
        for (Long id : collect) {
            flinkDataSyncService.unPublish(id);

            StartOnlineKafkaTaskParam startOnlineKafkaTaskParam = new StartOnlineKafkaTaskParam();
            startOnlineKafkaTaskParam.setTaskId(id);
            flinkDataSyncService.publish(startOnlineKafkaTaskParam);
        }
    }
}
